package com.shujia.spark.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo01WordCntOnStream {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      // 因为流处理需要一直接收数据，所以需要一直占用一个线程，
      // 所以至少需要2个及以上的线程才能保证实时任务正常运行
      .master("local[2]")
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .getOrCreate()

    // 构建SparkStreaming环境
    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    /**
     * 在master节点启动Socket服务：
     * nc -lk master 8888
     * 如果没有nc命令：yum install nc
     */
    val textDS: DStream[String] = ssc.socketTextStream("master", 8888)

    // 每隔指定的5s 就会打印前10条数据
    //    textDS.print()
    textDS
      // 将每个批次转换成一个个RDD再对每一个RDD进行处理
      .foreachRDD(rdd => {
        rdd
          .flatMap(_.split(","))
          .map(word => (word, 1))
          .reduceByKey(_ + _)
          .foreach(println)
      })

    // 需要启动streaming的任务
    ssc.start()
    ssc.awaitTermination() // 等待任务停止
    ssc.stop()


  }

}
